use crate::abstractions::data::Data;
use crate::prelude;
use crate::protocol::task::TaskInitD;
use crate::protocol::task::TaskStatus;
use crate::protocol::task::TaskProgressD;
use crate::protocol::type_schema::TypeSchema;
use crate::services::manager;
use std::io::Write;
use std::sync::atomic::AtomicBool;
use std::sync::mpsc::{Receiver, Sender};
use std::fs::File;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;
use super::manager::TASK_MANAGER;
use super::stream::get_pipe_handle;
use crate::error::ServerError;
use log::warn;
use uuid;
use crate::protocol::data_schema::DataTrait;
use std::sync::mpsc::TryRecvError;

type TaskFunc = Arc<dyn Fn(Data) -> Result<Data, ServerError>>;

#[derive(Debug)]
pub struct Task {
    pub workflow_id: Option<u16>,
    pub id: String,
    pub remote_id: String,
    pub service_addr: String,
    pub service_name: String,
    pub name: String,
    pub input_type: TypeSchema,
    pub output_type: TypeSchema,
    pub status: TaskStatus,
    pub created_at: u64,
    pub completed_at: Option<u64>,
    pub error_message: Option<String>,
    pub progress_percentage: f64,
    pub progress_receiver: Option<(Receiver<String>, Arc<AtomicBool>)>,
    pub log_receiver: Option<(Receiver<String>, Arc<AtomicBool>)>,
    pub log_sender: Option<Sender<String>>,
    pub log_file_name: Option<String>,
    pub log_file: Option<File>,
    pub result_sender_addr: String,
    pub terminate_flag: Arc<Mutex<bool>>, // 手动终止
    pub end_flag: Arc<Mutex<bool>>, // 任务结束
    pub result: Option<Data>,
    pub session_key: String
}

impl Task {
    pub fn new(service_addr: String, service_name: String, name: String, input_type: TypeSchema, output_type: TypeSchema, session_key: String) -> Self {
        Task {
            workflow_id: None,
            id: uuid::Uuid::new_v4().to_string(),
            remote_id: "".to_string(),
            service_addr,
            service_name,
            name,
            input_type,
            output_type,
            status: TaskStatus::Created,
            created_at: prelude::now(),
            completed_at: None,
            error_message: None,
            progress_percentage: 0.0,
            progress_receiver: None,
            log_receiver: None,
            log_sender: None,
            log_file_name: None,
            log_file: None,
            result_sender_addr: "".to_string(),
            terminate_flag: Arc::new(Mutex::new(false)),
            end_flag: Arc::new(Mutex::new(false)),
            result: None,
            session_key: "".to_string()
        }
    }

    pub fn new_from_workflow(service_addr: String, service_name: String, name: String, input_type: TypeSchema, output_type: TypeSchema, session_key: String, workflow_port: u16) -> Self {
        Task {
            workflow_id: Some(workflow_port),
            ..Task::new(service_addr, service_name, name, input_type, output_type, session_key)
        }
    }

    pub async fn start(mut self, input: Data) -> Result<(), ServerError> {
        let instant = self.connect(input).await?;
        if instant {
            println!("Instant任务结果已返回: {} - {} - {}", self.service_name, self.name, self.id);
            return Ok(());
        }
        println!("任务已启动: {} - {} - {}", self.service_name, self.name, self.id);
        self.start_stream_operation_background().await?;
        Ok(())
    }
    
    pub async fn connect(&mut self, input: Data) -> Result<bool, ServerError> {
        let mut init_info: TaskInitD;
        if self.service_addr.starts_with("http") {
            let url = format!("{}/task/{}", self.service_addr, self.name);
            // insert session_key into header
            let mut headers = reqwest::header::HeaderMap::new();
            headers.insert("session_key", reqwest::header::HeaderValue::from_str(&self.session_key).unwrap());
            let body = input.to_json_less();
            match reqwest::Client::new().post(&url).json(&body).headers(headers).send().await {
                Ok(response) => {
                    if response.status() != 200 {
                        return Err(ServerError::TaskError(format!("启动外部任务 {} 失败: {}", url, response.text().await.unwrap_or_default())))
                    }
                    let s = response.text().await.map_err(|_|ServerError::TaskErrorStr("外部服务任务启动无文本"))?;
                    init_info = serde_json::from_str(&s).map_err(|_|ServerError::TaskError(format!("外部服务任务启动响应解析失败: {}", s)))?;
                    
                }
                Err(e) => {
                    // change service status
                    return Err(ServerError::ServiceCannotReach)
                }
            }
        } else {
            let url = format!("/task/{}", self.name);
            let context = zmq::Context::new();
            let socket = context.socket(zmq::REQ).unwrap();
            socket.connect(&url).unwrap();
            let mut msg = zmq::Message::new();
            let payload = url + "#" + &serde_json::to_string(&input.to_json_less()).unwrap();
            socket.send(payload.as_bytes(), 0).unwrap();
            socket.recv(&mut msg, 0).unwrap();
            init_info = serde_json::from_str(&msg.as_str().unwrap()).unwrap();
        }
        self.remote_id = init_info.task_id.clone();
        if let Some(instant_result) = init_info.instant_result {
            match init_info.status {
                Some(status) => {
                    match status {
                        TaskStatus::Failed => {
                            self.error_message = init_info.error_message;
                            self.end_set_status(status);
                        },
                        TaskStatus::Completed => {
                            self.result = Some(Data::from_json_less(instant_result, &self.output_type)?);
                            self.end_set_status(TaskStatus::ResultReceived);
                        },
                        _ => unreachable!()
                    }
                },
                None => {
                    self.end_set_status(TaskStatus::ResultReceived);
                }
            }
            return Ok(true);
        }
        // add service base_addr
        if self.service_addr.starts_with("http") {
            init_info.progress_pipe_addr = self.service_addr.clone() + &init_info.progress_pipe_addr;
            if let Some(log_addr) = init_info.log_pipe_addr {
                init_info.log_pipe_addr = Some(self.service_addr.clone() + &log_addr);
            }
            let handles = get_pipe_handle(init_info.progress_pipe_addr)?;
            self.progress_receiver = Some(handles);
            if let Some(log_addr) = init_info.log_pipe_addr {
                let handles = get_pipe_handle(log_addr)?;
                self.log_receiver = Some(handles);
            }
            self.result_sender_addr = self.service_addr.clone() + &init_info.result_addr;
        } else {
            init_info.progress_pipe_addr = "tcp://localhost:".to_string() + &init_info.progress_pipe_addr[1..];
            if let Some(log_addr) = init_info.log_pipe_addr {
                init_info.log_pipe_addr = Some("tcp://localhost:".to_string() + &log_addr[1..]);
            }
            let handles = get_pipe_handle(init_info.progress_pipe_addr)?;
            self.progress_receiver = Some(handles);
            if let Some(log_addr) = init_info.log_pipe_addr {
                let handles = get_pipe_handle(log_addr)?;
                self.log_receiver = Some(handles);
            }
            self.result_sender_addr = init_info.result_addr;
        }
        Ok(false)
    }

    async fn start_stream_operation_background(mut self) -> Result<(), ServerError> {
        let mut progress_receiver = self.progress_receiver.take();
        let mut log_receiver = self.log_receiver.take();
        let task_id = self.id.clone();
        let terminate_flag = self.terminate_flag.clone();
        let info_str = format!("{} - {} - {}", self.service_name, self.name, self.id);
        let mut manager = TASK_MANAGER.get().unwrap().lock().await;
        manager.add_task(self)?;
        drop(manager);
        let handle = std::thread::spawn(move || {
            let rt = tokio::runtime::Runtime::new().unwrap();
            let res = rt.block_on(async move {
                let mut exit_status = TaskStatus::Running;
                'main_loop: loop {
                    // manual stop
                    if *terminate_flag.lock().unwrap() {
                        println!("手动终止任务: {}", task_id);
                        exit_status = TaskStatus::Stopped;
                        break 'main_loop;
                    }
                    if let Some((ref mut receiver, _)) = progress_receiver {
                        match receiver.try_recv() {
                            Ok(progress_str) => {
                                if progress_str == "Closed" {
                                    exit_status = TaskStatus::Closed;
                                    break 'main_loop;
                                }
                                match serde_json::from_str::<TaskProgressD>(&progress_str) {
                                    Ok(progress) => {
                                        let mut manager = TASK_MANAGER.get().unwrap().lock().await;
                                        let task = match manager.get_task_mut(&task_id) {
                                            Some(v) => v,
                                            None => panic!("{}", format!("task not found: {}", task_id))
                                        };
                                        task.update_progress(progress.status, progress.percentage, progress.errmsg);
                                        drop(manager);
                                        match progress.status {
                                            TaskStatus::Completed | TaskStatus::Failed | TaskStatus::Stopped => {
                                                exit_status = progress.status;
                                                break 'main_loop;
                                            },
                                            TaskStatus::Running | TaskStatus::Created => {},
                                            other => {
                                                println!("远程不应该返回此状态：{:?}", other);
                                            }
                                        }
                                    },
                                    Err(e) => {
                                        println!("解析进度信息失败: {}, 原始数据: {}", e, progress_str);
                                        break 'main_loop;
                                    }
                                }
                            },
                            Err(TryRecvError::Disconnected) => {println!("task - background_thread - channel disconnected"); break},
                            _ => {}
                        }
                    }
                    if let Some((ref mut receiver, _)) = log_receiver {
                        if let Ok(log) = receiver.try_recv() {
                            // println!("task - background_thread - 任务日志: {}, ID: {}", log, task_id);
                            // let mut manager = TASK_MANAGER.get().unwrap().lock().await;
                            // let task = manager.get_task_mut(&task_id).unwrap();
                            // // task.log(log);
                            // drop(manager);
                        }
                    }
                    std::thread::sleep(std::time::Duration::from_millis(200));
                }
                if let Some((_, switch)) = progress_receiver {
                    switch.store(false, std::sync::atomic::Ordering::Relaxed);
                }
                if let Some((_, switch)) = log_receiver {
                    switch.store(false, std::sync::atomic::Ordering::Relaxed);
                }
                println!("任务结束: {}, 退出状态: {:?}", info_str, exit_status);
                let mut manager = TASK_MANAGER.get().unwrap().lock().await;
                manager.complete_task(task_id, exit_status)?;
                drop(manager);
                Result::<(), ServerError>::Ok(())
            });
            res
        });
        Ok(())
    }

    pub fn terminate(&self) -> Result<(), ServerError> {
        let mut terminate = self.terminate_flag.lock().unwrap();
        *terminate = true;
        let service_addr = self.service_addr.clone();
        let remote_task_id = self.remote_id.clone();
        let task_name = self.name.clone();
        let handle = std::thread::spawn(move || {
            let rt = tokio::runtime::Runtime::new().unwrap();
            let res = rt.block_on(async move {
                let url = format!("{}/task/{}/terminate/{}", service_addr, task_name, remote_task_id);
                let client = reqwest::Client::new();
                client.post(url).send().await.map_err(|_| ServerError::TaskErrorStr("终止任务失败"))
            });
            res
        });
        let res = handle.join().unwrap();
        match res {
            Ok(_) => Ok(()),
            Err(e) => Err(e)
        }
    }

    pub fn set_status(&mut self, status: TaskStatus) {
        self.status = status
    }

    pub fn end_set_status(&mut self, status: TaskStatus) {
        let old_status = self.status;
        self.status = status;
        if self.completed_at.is_some() {
            warn!("重复设置任务完成: {}: {:?}和{:?}", self.id, old_status, status);
        }
        self.completed_at = Some(prelude::now());
    }

    pub fn update_progress(&mut self, status: TaskStatus, progress: f64, error_message: Option<String>) {
        self.progress_percentage = progress.min(100.0);
        self.status = status;
        self.error_message = error_message;
    }

    pub fn log(&mut self, message: String) {
        if let Some(sender) = &self.log_sender {
            let _ = sender.send(message.clone());
        }
        if let Some(log_file) = &mut self.log_file {
            log_file.write_all(message.as_bytes()).unwrap();
        }
    }
}

pub async fn task_get_result(id: String) -> Result<Data, ServerError> {
    let manager = TASK_MANAGER.get().unwrap().lock().await;
    let task = manager.get_task(&id).unwrap();
    let service_addr = task.service_addr.clone();
    let task_id = task.id.clone();
    let task_name = task.name.clone();
    let output_type = task.output_type.clone();
    let task_info = format!("{} - {} - {}", task.service_name, task.name, task.id);
    let url = task.result_sender_addr.clone();
    drop(manager);
    // println!("getting result: {}", url);
    let client = reqwest::Client::new();
    match client.get(url).timeout(Duration::from_millis(1000)).send().await {
        Ok(response) => {
            let str_result = response.text().await.unwrap();
            let json_result: serde_json::Value = serde_json::from_str(&str_result).map_err(|_| ServerError::TaskErrorStr("结果不是json格式"))?;
            let result = Data::from_json_less(json_result, &output_type)?;
            // println!("qualify: {:?} by {:?}", result, output_type);
            if !result.qualify(&output_type) {
                println!("结果类型不匹配: {}", task_info);
                return Err(ServerError::TaskErrorStr("结果类型不匹配"));
            }
            println!("获取结果成功: {}", task_info);
            Ok(result)
        },
        Err(e) => {
            return Err(ServerError::TaskError(format!("获取结果失败: {}", e)));
        }
    }
}
